package com.mq.produce.simple.v2;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @Classname CustomerProduce
 * @Description 生产者发送消息应答机制
 * @Version 1.0.0
 * @Date 2025/3/7 9:53
 * @Created by ncs
 */
public class CustomerProduceAcks {

    static String kafkaServer = "192.168.125.100:9093,192.168.125.100:9094,192.168.125.100:9095";

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //acks 0:leader收到消息就应答生产者,1:leader落盘以后应答生产者;-1(all):生产者落盘,isr的broker都收到以后应答生产者。默认等于-1(all)
        properties.put(ProducerConfig.ACKS_CONFIG, "1");
        //配置重试次数,默认值是int的最大值,当生产者没有收到broker的应答消息的时候会触发重试，默认int的最大值,
        //基于这种情况,isr收到消息应答给生产者的时候网络原因会导致生产者重复发送消息
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        for (int i = 0; i < 5; i++) {
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("first-topic", "first-message" + i);
            kafkaProducer.send(producerRecord);
        }
        kafkaProducer.close();
    }
}
